package com.shujia.flink.sql

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo6TopNEventTime {

  def main(args: Array[String]): Unit = {


    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    bsEnv.setParallelism(1)

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //指定计划器为blink
      .inStreamingMode() //指定流模式
      .build()
    /**
      *
      * 构建flink table环境
      *
      */
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

/*

java,20180503122110
java,20180503122111
java,20180503122112
java,20180503122113
spark,20180503122110
spark,20180503122111
spark,20180503122112
hive,20180503122110
hive,20180503122111
hive,20180503122141

 */

    bsTableEnv.executeSql(
      """
        |CREATE TABLE words_source (
        |  word STRING,
        |  data_time STRING,
        |  ts as TO_TIMESTAMP(data_time,'yyyyMMddHHmmss'),
        |  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
        |) WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'words',
        |  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        |  'properties.group.id' = 'asdasd',
        |  'format' = 'csv',
        |  'csv.ignore-parse-errors' = 'true'
        |)
        |
      """.stripMargin)


    bsTableEnv.from("words_source").printSchema()

    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE top_sink (
        |  hop_end STRING,
        |  r bigint,
        |  word STRING,
        |  c bigint,
        |  PRIMARY KEY (hop_end,r) NOT ENFORCED
        |) WITH (
        |  'connector' = 'jdbc',
        |  'url' = 'jdbc:mysql://master:3306/student?useUnicode=true&characterEncoding=UTF-8',
        |  'table-name' = 'top',
        |  'username' = 'root',
        |  'password' = '123456'
        |)
        |
      """.stripMargin)

    /**
      * 统计最近15秒单词数量最多的前两个单词，每隔两秒统计一次
      *
      */


    bsTableEnv.executeSql(
      """
        |
        insert into top_sink
        |select DATE_FORMAT(hop_end,'yyyyMMddHHmmss') as hop_end ,r ,  word,  c  from
        |( select word,hop_end,c ,row_number() over(partition by hop_end order by c desc ) as r  from
        |(select word,HOP_END(ts, INTERVAL  '2' SECOND ,INTERVAL  '15' SECOND) as hop_end ,count(1) c
        |from words_source
        |group by word,HOP(ts, INTERVAL  '2' SECOND ,INTERVAL  '15' SECOND)
        |) as a
        |) as b
        |where r <= 2
        |
      """.stripMargin)


  }

}
